package com.example.sendmessage.client;

import com.example.sendmessage.common.*;
import com.example.sendmessage.server.core.BasicProperties;
import com.example.sendmessage.server.core.ExchangeType;

import java.io.IOException;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

public class Channel {
    // 频道ID，当前channel属于哪个连接
    private String channelId;
    // 存储后续客户端收到的服务器响应
    private Connection connection;
    // 存储后续客户端收到的服务器响应
    private ConcurrentHashMap<String, BasicReturn> basicReturnMap = new ConcurrentHashMap<>();

    private Consumer consumer = null;
    public Channel(String channelId, Connection connection) {
        this.channelId = channelId;
        this.connection = connection;
    }

    /**
     * 利用此方法创建一个channel，并发送给服务器
     * @return 返回响应的结果
     */
    public boolean createChannel() throws IOException {
        BasicArgument basicArgument = new BasicAckArgument();
        basicArgument.setChannelId(channelId);
        basicArgument.setRid(generateRid());
        byte[] payload = BinaryTool.toBytes(basicArgument);
        Request request = new Request();
        request.setType(0x1);
        request.setLength(payload.length);
        request.setPayload(payload);
        // 将请求发送给服务器
        connection.writeRequest(request);
        // 等待服务器返回响应
        BasicReturn basicReturn = waitResponse(basicArgument.getRid());
        return basicReturn.isOk();
    }

    private String generateRid() {
        // 生成一个请求id，该id为请求和响应的唯一对应
        return "R-" + UUID.randomUUID();
    }

    private BasicReturn waitResponse(String rid) {
        // 等待服务器返回响应
        BasicReturn basicReturn = null;
        while ((basicReturn = basicReturnMap.get(rid)) == null) {
            // 等待服务器的响应，如果响应是空就阻塞等待
            synchronized (this) {
                try {
                    wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
        // 服务器响应来了就将rid从map中移除
        basicReturnMap.remove(rid);
        return basicReturn;
    }


    /**
     * 关闭channel连接
     * @return 服务器响应结果
     * @throws IOException  字节流转化异常
     */
    public boolean close() throws IOException {
        // 发送一个关闭请求给服务器
        BasicArgument basicArgument = new BasicAckArgument();
        basicArgument.setChannelId(channelId);
        basicArgument.setRid(generateRid());
        byte[] payload = BinaryTool.toBytes(basicArgument);
        Request request = new Request();
        request.setType(0x2);
        request.setLength(payload.length);
        request.setPayload(payload);
        connection.writeRequest(request);
        // 等待服务器响应
        BasicReturn basicReturn = waitResponse(basicArgument.getRid());
        return basicReturn.isOk();
    }

    /**
     * 发送创建交换机的请求给服务器
     * @param exchangeName 交换机名字
     * @param type 交换机类型
     * @param durable 交换机是否持久化
     * @param autoDelete 交换机是否自动删除
     * @param arguments 交换机其他参数
     * @return 服务器返回交换创建得响应
     * @throws IOException 字节流异常
     */
    public boolean exchangeDeclare(String exchangeName, ExchangeType type, boolean durable,
                                   boolean autoDelete, Map<String, Object> arguments) throws IOException {
        ExchangeDeclareArgument argument = new ExchangeDeclareArgument();
        argument.setChannelId(channelId);
        argument.setRid(generateRid());
        argument.setExchangeName(exchangeName);
        argument.setExchangeType(type);
        argument.setDurable(durable);
        argument.setAutoDelete(autoDelete);
        argument.setArguments(arguments);
        byte[] payload = BinaryTool.toBytes(argument);
        Request request = new Request();
        request.setType(0x3);
        request.setLength(payload.length);
        request.setPayload(payload);
        // 请求发送个服务器
        connection.writeRequest(request);
        BasicReturn basicReturn = waitResponse(argument.getRid());
        return basicReturn.isOk();
    }

    /**
     * 删除交换机
     * @param exchangeName 删除交换机的名字
     * @return 响应结果
     * @throws IOException 字节数组转换异常
     */
    public boolean exchangeDelete(String exchangeName) throws IOException {
        ExchangeDeleteArgument argument = new ExchangeDeleteArgument();
        argument.setChannelId(channelId);
        argument.setRid(generateRid());
        argument.setExchangeName(exchangeName);
        byte[] payload = BinaryTool.toBytes(argument);
        Request request = new Request();
        request.setType(0x4);
        request.setLength(payload.length);
        request.setPayload(payload);
        // 请求发送个服务器
        connection.writeRequest(request);
        BasicReturn basicReturn = waitResponse(argument.getRid());
        return basicReturn.isOk();
    }

    /**
     * 创建队列
     * @param queueName 创建的队列名
     * @param durable 是否持久化
     * @param exclusive 是否专用
     * @param autoDelete 是否自动删除
     * @param arguments 其他参数
     * @return 服务器响应结果
     * @throws IOException 字节数组转换异常
     */
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive,
                                boolean autoDelete, Map<String, Object> arguments) throws IOException {
        // 请求消息体
        QueueDeclareArgument argument = new QueueDeclareArgument();
        argument.setChannelId(channelId);
        argument.setRid(generateRid());
        argument.setQueueName(queueName);
        argument.setDurable(durable);
        argument.setExclusive(exclusive);
        argument.setAutoDelete(autoDelete);
        argument.setArguments(arguments);
        byte[] payload = BinaryTool.toBytes(argument);
        // 构造请求
        Request request = new Request();
        request.setType(0x5);
        request.setLength(payload.length);
        request.setPayload(payload);
        // 请求发送个服务器
        connection.writeRequest(request);
        // 等待服务器返回响应
        BasicReturn basicReturn = waitResponse(argument.getRid());
        return basicReturn.isOk();
    }

    /**
     *  删除队列
     * @param queueName 删除的队列名
     * @return 服务器返回的响应
     * @throws IOException 字节数组转换异常
     */
    public boolean queueDelete(String queueName) throws IOException {
        QueueDeleteArgument argument = new QueueDeleteArgument();
        argument.setChannelId(channelId);
        argument.setRid(generateRid());
        argument.setQueueName(queueName);
        byte[] payload = BinaryTool.toBytes(argument);
        Request request = new Request();
        request.setType(0x6);
        request.setLength(payload.length);
        request.setPayload(payload);
        connection.writeRequest(request);
        BasicReturn basicReturn = waitResponse(argument.getRid());
        return basicReturn.isOk();
    }

    /**
     * 交换机绑定队列
     * @param exchangeName 绑定的交换机名
     * @param queueName 绑定的队列名
     * @param bindingKey 绑定关键字
     * @return 服务器返回绑定的结果
     * @throws IOException 字节数据异常
     */
    public boolean queueBind(String queueName, String exchangeName, String bindingKey) throws IOException {
        QueueBindArgument argument = new QueueBindArgument();
        argument.setChannelId(channelId);
        argument.setRid(generateRid());
        argument.setExchangeName(exchangeName);
        argument.setQueueName(queueName);
        argument.setBindingKey(bindingKey);
        byte[] payload = BinaryTool.toBytes(argument);
        Request request = new Request();
        request.setType(0x7);
        request.setLength(payload.length);
        request.setPayload(payload);
        connection.writeRequest(request);
        BasicReturn basicReturn = waitResponse(argument.getRid());
        return basicReturn.isOk();
    }

    /**
     * 解除交换机与队列的绑定
     * @param queueName 解绑的队列名
     * @param exchangeName 解绑的交换机名
     * @return 解除绑定的结果
     * @throws IOException 字节数组处理异常
     */
    public boolean queueUnbind(String queueName, String exchangeName) throws IOException {
        QueueUnbindArgument argument = new QueueUnbindArgument();
        argument.setRid(generateRid());
        argument.setChannelId(channelId);
        argument.setQueueName(queueName);
        argument.setExchangeName(exchangeName);
        byte[] payload = BinaryTool.toBytes(argument);

        Request request = new Request();
        request.setType(0x8);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturn basicReturns = waitResponse(argument.getRid());
        return basicReturns.isOk();
    }

    /**
     * 消息发送
     * @param exchangeName 发送到的交换机
     * @param routingKey 核对关键词
     * @param basicProperties 消息关键部分
     * @param body 消息内容
     * @return 消息发送的结果
     * @throws IOException 字节处理异常
     */
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body) throws IOException {
        BasicPublishArgument arguments = new BasicPublishArgument();
        arguments.setChannelId(channelId);
        arguments.setRid(generateRid());
        arguments.setExchangeName(exchangeName);
        arguments.setRoutingKey(routingKey);
        arguments.setBasicProperties(basicProperties);
        arguments.setBody(body);
        byte[] payload = BinaryTool.toBytes(arguments);

        Request request = new Request();
        request.setType(0x9);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturn basicReturns = waitResponse(arguments.getRid());
        return basicReturns.isOk();
    }

    /**
     * 订阅队列的消息
     * @param queueName 队列名字
     * @param autoAck 是否自动响应
     * @param consumer 回调
     * @return 订阅队列消息的结果
     * @throws MqException 重复设置回调
     * @throws IOException 字节数组处理异常
     */
    public boolean basicConsume(String queueName, boolean autoAck, Consumer consumer) throws MqException, IOException {
        // 先设置回调.
        if (this.consumer != null) {
            throw new MqException("该 channel 已经设置过消费消息的回调了, 不能重复设置!");
        }
        this.consumer = consumer;

        BasicConsumeArgument arguments = new BasicConsumeArgument();
        arguments.setChannelId(channelId);
        arguments.setRid(generateRid());
        arguments.setQueueName(queueName);
        arguments.setAutoAck(autoAck);
        arguments.setConsumerTag(channelId);
        byte[] payload = BinaryTool.toBytes(arguments);

        Request request = new Request();
        request.setType(0xa);
        request.setLength(payload.length);
        request.setPayload(payload);
        connection.writeRequest(request);
        BasicReturn basicReturn = waitResponse(arguments.getRid());
        return basicReturn.isOk();
    }

    /**
     * 消息手动应答
     * @param queueName 应答的队列名字
     * @param messageId 应答的消息id
     * @return 消息手动应答的响应结果
     * @throws IOException 字节数组处理异常
     */
    public boolean basicAck(String queueName, String messageId) throws IOException {
        BasicAckArgument argument = new BasicAckArgument();
        argument.setChannelId(channelId);
        argument.setRid(generateRid());
        argument.setQueueName(queueName);
        argument.setMessageId(messageId);
        byte[] payload = BinaryTool.toBytes(argument);

        Request request = new Request();
        request.setType(0xb);
        request.setLength(payload.length);
        request.setPayload(payload);

        connection.writeRequest(request);
        BasicReturn basicReturn = waitResponse(argument.getRid());
        return basicReturn.isOk();
    }

    public String getChannelId() {
        return channelId;
    }

    public void setChannelId(String channelId) {
        this.channelId = channelId;
    }

    public Connection getConnection() {
        return connection;
    }

    public void setConnection(Connection connection) {
        this.connection = connection;
    }

    public ConcurrentHashMap<String, BasicReturn> getBasicReturnMap() {
        return basicReturnMap;
    }

    public void setBasicReturnMap(ConcurrentHashMap<String, BasicReturn> basicReturnMap) {
        this.basicReturnMap = basicReturnMap;
    }

    public Consumer getConsumer() {
        return consumer;
    }

    public void setConsumer(Consumer consumer) {
        this.consumer = consumer;
    }
}
